package list;


import sun.misc.Unsafe;

import java.lang.reflect.Field;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.LockSupport;

/**
 * @user 26233
 * @create 2020-07-25 16:26:03
 * @description 队列同步器
 **/

public class MyAbstractQueuedSynchronizer {

    // 同步状态
    private volatile int state;

    private static Unsafe unsafe;

    private Thread exclusiveOwnerThread = null;

    // 字节长度填充？（和底层CAS操作相关？）
    private static final long stateOffset;
    private static final long tailOffset;
    private static final long headOffset;
    private static final long waitStatusOffset;
    static {
        Field f = null;
        try {
            f = Unsafe.class.getDeclaredField("theUnsafe");
            f.setAccessible(true);
            unsafe = (Unsafe) f.get(null);
        } catch (NoSuchFieldException e) {
            e.printStackTrace();
        } catch (IllegalAccessException e) {
            e.printStackTrace();
        }
        try {
            stateOffset = unsafe.objectFieldOffset
                    (MyAbstractQueuedSynchronizer.class.getDeclaredField("state"));
            tailOffset = unsafe.objectFieldOffset
                    (MyAbstractQueuedSynchronizer.class.getDeclaredField("tail"));
            headOffset = unsafe.objectFieldOffset
                    (MyAbstractQueuedSynchronizer.class.getDeclaredField("head"));
            waitStatusOffset = unsafe.objectFieldOffset
                    (AQSNode.class.getDeclaredField("waitStatus"));
        } catch (Exception ex) { throw new Error(ex); }
    }

    public void setState(int state){
        this.state = state;
    }

    public int getState() {
        return state;
    }

    public boolean compareAndSetState(int expect, int update){
        return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
    }

    public void setExclusiveOwnerThread(Thread exclusiveOwnerThread) {
        this.exclusiveOwnerThread = exclusiveOwnerThread;
    }

    // -------------------------------------
    private AQSNode head;

    private AQSNode tail;


    static class AQSNode {
        AQSNode pre;
        AQSNode next;
        Thread thread;
        // ---------------------------------
        int waitStatus;
        AQSNode nextWaiter;
        static AQSNode EXCLUSIVE = null;

        final static int SIGNAL = -1;

        public AQSNode(){}

        public AQSNode(Thread thread, AQSNode mode){
            this.thread = thread;
            this.nextWaiter = mode;
        }

        public AQSNode predecessor(){
            AQSNode n = pre;
            if (n == null){
                throw new NullPointerException();
            }
            return n;
        }
    }
    // 获取同步状态
    public void acquire(int arg){
        // 获取同步状态失败
        if(!tryAcquire(arg) && acquireQueued(addWaiter(AQSNode.EXCLUSIVE), arg)){
            // 获取同步状态时报，中断获取的线程
            Thread.currentThread().interrupt();
        }
        // 获取成功，执行临界区代码...
    }

    // 获取同步状态
    public boolean tryAcquire(int arg){
        if(compareAndSetState(0, arg)){
            setExclusiveOwnerThread(Thread.currentThread());
            return true;
        }
        return false;
    }

    private boolean release(int arg) throws IllegalAccessException {

        if (tryRelease(arg)){
            AQSNode h = head;
            if (h != null && h.waitStatus != 0){
                unparkSuccessor(h);
            }
            return true;
        }
        return false;
    }

    // 唤醒后继节点中的线程
    private void unparkSuccessor(AQSNode node) {
        int ws = node.waitStatus;
        if (ws < 0)
            compareAndSetWaitStatus(node, ws, 0);
        AQSNode s = node.next;
        if (s == null || s.waitStatus > 0) {
            s = null;
            for (AQSNode t = tail; t != null && t != node; t = t.pre)
                if (t.waitStatus <= 0)
                    s = t;
        }
        if (s != null)
            LockSupport.unpark(s.thread);
    }

    private boolean tryRelease(int arg) throws IllegalAccessException {
        if (getState() == 0) throw new IllegalAccessException();
        setExclusiveOwnerThread(null);
        setState(0);
        return true;
    }

    public void setHead(AQSNode node){
        head = node;
        node.pre = null;
        node.thread = null;
    }

    /**
     * 循环CAS尝试获取同步状态
     * @param node new tail of setting successful
     * @param arg
     * @return
     */
    public boolean acquireQueued(AQSNode node, int arg){
        boolean failed = true;
        try{
            boolean interrupted = false;
            for (;;){
                // 本节点的前驱节点
                AQSNode p = node.predecessor();
                // 本节点的前驱是head节点，并且成功获取到同步状态
                if(p == head && tryAcquire(arg)){
                    setHead(node);// 把自己设置为头结点
                    p.next = null;
                    failed = false;
                    // 方法返回
                    return interrupted;
                }
                // p=null ?
                // 线程挂起
                if(shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()){
                    interrupted = true;
                }

            }
        }finally {
            if (failed)
                cancelAcquire(node);
        }
    }

    private boolean parkAndCheckInterrupt() {
        LockSupport.park(this);
        return Thread.interrupted();
    }

    /**
     *
     * @param pre predecessor of node
     * @param node
     * @return
     */
    private boolean shouldParkAfterFailedAcquire(AQSNode pre, AQSNode node) {
        int ws = pre.waitStatus;
        // 前驱已调用release,释放同步状态
        if(ws == AQSNode.SIGNAL){
            return true;
        }
        //
        if(ws > 0){
            // 将前驱节点同步状态为异常从链表中去掉
            do{
                node.pre = pre = pre.pre;
            }while (pre.waitStatus > 0);
            pre.next = node;
        }else{
            // 请求释放
            compareAndSetWaitStatus(pre, ws, AQSNode.SIGNAL);
        }
        return false;
    }

    private boolean compareAndSetWaitStatus(AQSNode pre, int expect, int update) {
        return unsafe.compareAndSwapInt(pre, waitStatusOffset, expect, update);
    }

    public void cancelAcquire(AQSNode node){

    }

    // 设置尾节点
    private boolean compareAndSetTail(AQSNode expect, AQSNode update){
        return unsafe.compareAndSwapObject(this, tailOffset, expect, update);
    }
    // 设置尾节点
    private boolean compareAndSetHead(AQSNode expect, AQSNode update){
        return unsafe.compareAndSwapObject(this, headOffset, expect, update);
    }

    // 把节点设置为尾结点
    private AQSNode addWaiter(AQSNode mode){
        // 构造当前线程的节点
        AQSNode node = new AQSNode(Thread.currentThread(), mode);
        AQSNode pred = tail;
        // 链表中有非空节点，快速尝试
        if(pred != null){
            node.pre = tail;
            // 把新节点设置为尾结点，原子操作
            if(compareAndSetTail(pred, node)){
                pred.next = node;
                return node;
            }
        }
        // 空链表
        // CAS设置尾结点失败，自旋循环设置
        enq(node);
        return node;
    }
    // 循环CAS尝试将自己设置为尾结点

    /**
     *
     * @param node
     * @return tail of predecessor, 返回值不重要
     */
    private AQSNode enq(AQSNode node){
        for (;;){
            AQSNode t = tail;
            if(t == null){// 空链表
                if(compareAndSetHead(null, new AQSNode())){
                    tail = head;
                }
            }else{
                node.pre = t;
                if(compareAndSetTail(t, node)){
                    t.next = node;
                    // predecessor
                    return t;
                }
            }
        }
    }

    public static void main(String[] args) throws Exception {

        MyAbstractQueuedSynchronizer m = new MyAbstractQueuedSynchronizer();

        new Thread(new Runnable() {
            @Override
            public void run() {
                m.acquire(1);
                System.out.println(Thread.currentThread() + " get m");
                try {
                    TimeUnit.SECONDS.sleep(10);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println(Thread.currentThread() + " done");
                try {
                    m.release(1);
                } catch (IllegalAccessException e) {
                    e.printStackTrace();
                }
            }
        }, "my_thread-1").start();

        new Thread(new Runnable() {
            @Override
            public void run() {
                m.acquire(1);
                System.out.println(Thread.currentThread() + " get m");
                try {
                    TimeUnit.SECONDS.sleep(10);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println(Thread.currentThread() + " done");
                try {
                    m.release(1);
                } catch (IllegalAccessException e) {
                    e.printStackTrace();
                }
            }
        }, "my_thread-2").start();

        new Thread(new Runnable() {
            @Override
            public void run() {
                m.acquire(1);
                System.out.println(Thread.currentThread() + " get m");
                try {
                    TimeUnit.SECONDS.sleep(10);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println(Thread.currentThread() + " done");
                try {
                    m.release(1);
                } catch (IllegalAccessException e) {
                    e.printStackTrace();
                }
            }
        }, "my_thread-3").start();

        try {
            TimeUnit.SECONDS.sleep(2);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("hang on");

        // AbstractQueuedSynchronizer
    }
}
